package com.ry.flink.datagen;
import com.ry.flink.utils.Constants;
import com.ry.flink.utils.KafkaProducerUtils;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 *  生成client_log数据
 */
public class MockData {
    public static void main(String[] args) throws InterruptedException {
//        mockDataIndicators1();
        mockDataIndicators2();
    }
    private static void mockDataIndicators1() {
        Producer<String, String> kafkaProducer = KafkaProducerUtils.getProducer();
        while (true) {
            String message = "{\"datetime\":\""+MockDataUtils.getCurrentTime()+"\",\"countryID\":\""+MockDataUtils.getCountryID()+"\",\"data\":[{\"type\":\""+MockDataUtils.getRandomType()+"\",\"artifact_code\":"+MockDataUtils.getArtifactCode()+",\"sku_codechar\":\""+MockDataUtils.getSkuCodeChar()+"\"},{\"type\":\""+MockDataUtils.getRandomType()+"\",\"artifact_code\":"+MockDataUtils.getArtifactCode()+",\"sku_codechar\":\""+MockDataUtils.getSkuCodeChar()+"\"}]}";
            System.out.println(message);
            ProducerRecord<String, String> record = new ProducerRecord<>(Constants.INDICATORS1_LOG,message);
            kafkaProducer.send(record);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private static void mockDataIndicators2() {
        Producer<String, String> kafkaProducer = KafkaProducerUtils.getProducer();
        while (true) {
            String message = "{\"datetime\":\""+MockDataUtils.getCurrentTime()+"\",\"type\":\""+MockDataUtils.getRandomType()+"\",\"check_username\":\""+MockDataUtils.getChineseName()+"\",\"area\":\""+MockDataUtils.getRandomArea()+"\"}";
            System.out.println(message);
            ProducerRecord<String, String> record = new ProducerRecord<>(Constants.INDICATORS2_LOG,message);
            kafkaProducer.send(record);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}
